package com.myflink.day05;

import com.project.capture5.bean.OrderLog;
import com.project.capture5.bean.ReceiptLog;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

/**
 * @author Shelly An
 * @create 2020/9/19 10:21
 * 实时对账监控 考虑数据乱序、数据迟到的场景
 */
public class ProcessFunction_CoProcessFuction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        SingleOutputStreamOperator<OrderLog> orderLogDS = env.readTextFile("Data/OrderLog.csv")
                .map(new MapFunction<String, OrderLog>() {
                    @Override
                    public OrderLog map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new OrderLog(Long.valueOf(datas[0]),
                                datas[1],
                                datas[2],
                                Long.valueOf(datas[3])
                        );
                    }
                });

        SingleOutputStreamOperator<ReceiptLog> txDS = env.readTextFile("Data/ReceiptLog.csv")
                .map(new MapFunction<String, ReceiptLog>() {
                    @Override
                    public ReceiptLog map(String value) throws Exception {
                        String[] datas = value.split(",");
                        return new ReceiptLog(datas[0], datas[1], Long.valueOf(datas[2]));
                    }
                });

        //两条流connect的时候，会做keyBy，是为了要匹配的数据到一起。先keyBy再connect
        ConnectedStreams<OrderLog, ReceiptLog> orderAndTxCS =
                //数据流1keyBy
                orderLogDS.keyBy(OrderLog::getTxId)
                        //连接keyBy后的数据流1  和 keyBy后的数据流2
                .connect(
                        //数据流2keyBy
                        txDS.keyBy(ReceiptLog::getTxId)
                );



        //2. 处理数据，实时对账  监控
        // 两条流 connect 起来，通过txId 做一个匹配，匹配上就是对账成功
        // 对于同一笔订单的交易来说，业务系统和交易系统的数据，哪个先来是不一定的
        SingleOutputStreamOperator<String> resultDS = orderAndTxCS.process(
                new CoProcessFunction<OrderLog, ReceiptLog, String>() {

                    //考虑什么时候回收  1. 对账成功 （已实现） 2. 订单支付超过半个小时，已失效 （未实现）
                    // 还有个问题：如果 支付数据延迟过久，怎么办？
                    //用来存放交易系统的数据
                    private Map<String, ReceiptLog> txMap = new HashMap<>();
                    //用来存放业务系统的数据
                    private Map<String, OrderLog> orderMap = new HashMap<>();

                    /**
                     * orderLogDS 来一条处理一条
                     * @param value 业务系统的数据
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processElement1(OrderLog value, Context ctx, Collector<String> out) throws Exception {
                        //进入这个方法，说明来的数据是业务系统的数据
                        //判断 交易数据来了没有。 通过交易码，获取保存的交易数据，如果为空，说明交易数据没来。  反之，交易数据已经来了即匹配上了。
                        ReceiptLog txEvent = txMap.get(value.getTxId());
                        if (txEvent == null) {
                            //1. 说明交易数据没来 把自己临时保存起来
                            orderMap.put(value.getTxId(), value);

                        } else {
                            //2. 说明交易数据来了  =》 对账成功
                            out.collect("订单" + value.getOrderId() + "对账成功");
                            //将保存的数据删掉
                            txMap.remove(value.getTxId());
                        }
                    }

                    /**
                     * txDS 来一条处理一条
                     * @param value 支付系统的数据
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processElement2(ReceiptLog value, Context ctx, Collector<String> out) throws Exception {
                        //逻辑同上
                        OrderLog orderEvent = orderMap.get(value.getTxId());
                        if (orderEvent == null) {
                            txMap.put(value.getTxId(), value);
                        } else {
                            out.collect("订单" + orderEvent.getOrderId() + "对账成功");
                            orderMap.remove(value.getTxId());
                        }
                    }
                }
        );

        resultDS.print();

        env.execute();
    }
}
